package a_aa_amainpackage.command_deal.dadanpin0403

import com.alibaba.fastjson.JSON
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark._

/*
*项目名: o2odata_commend
*文件名: FromEs
*创建者: hancongcong
*创建时间:2020/3/23 10:37
*描述: TODO
*/ object JdPddSnFromEs {

  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
//    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.29")
//    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    /**
      * 水果里的苹果、淘宝天猫：50050732
      * 盒马：tbCategoryId 126198303 126198306 126198308
      *
      * 苏宁：503926
      * 京东：13554
      * 拼多多：8269  2447  171  7718  苹果
      *
      * 禽蛋里的鸡蛋、淘宝天猫：50012385
      * 盒马：127526168 title含鸡蛋
      *
      * 京东:detailCategoryName 鸡蛋
      * 苏宁：subCategoryId 503945 title含鸡蛋
      * 拼多多:detailCategoryName 鸡蛋
      *
      * 蔬菜里的红薯、淘宝天猫：50050720
      * 盒马：127532130
      * 京东：detailCategoryName 番薯   地瓜/红薯
      * 苏宁：detailCategoryName 红薯
      * 拼多多：subCategoryId 8220
      *
      *
      * 鲜肉里的猪肉、淘宝天猫：50016811
      * 盒马：127530151
      * 127530152
      * 127530153
      *
      * 苏宁：detailCategoryName 猪肉 冰鲜猪肉
      * 京东：100210305 title含猪肉
      * 拼多多：detailCategoryName 生猪肉
      *
      *
      * 休闲食品里的坚果炒货；100210102
      * 牛奶乳品里的牛奶：100210501
      * 粮油调味的大米:100210203
      * 食品酒水里的茶：1002108
      * when categoryid ='50050732' then '苹果'
      * when categoryid ='50050720' then '红薯'
      * when thirdcategoryid='100210102' then '坚果炒货'
      * when thirdcategoryid='100210501' then '牛奶'
      * when secondcategoryid='1002108' then '茶'
      * --苹果：加力、爱妃、青苹果、蛇果、冰糖心、红富士、花牛、黄元帅、黄香蕉、嘎拉、金冠、红星、乔纳金、澳洲青果、国光
      * --茶：铁观音、普洱、绿茶、红茶、其他茶
      * --红薯：包括番薯、甘储、甘薯、朱薯、金薯、番茹、红山药、玉枕薯、山芋、地瓜、山药、甜薯、红薯、红苕、白薯、阿鹅、萌番薯。
      * --坚果炒货:夏威夷果、核桃、腰果、杏仁、巴旦木、松子、瓜子、腰果、开心果、核桃      ...以形式提：礼包、礼盒
      * --牛奶：常温/低温牛奶
      *
      * --苹果：加力、爱妃、青苹果、蛇果、冰糖心、红富士、花牛、黄元帅、黄香蕉、嘎拉、金冠、红星、乔纳金、澳洲青果、国光
      * --茶：铁观音、普洱、绿茶、红茶、其他茶
      * --红薯：包括番薯、甘储、甘薯、朱薯、金薯、番茹、红山药、玉枕薯、山芋、地瓜、山药、甜薯、红薯、红苕、白薯、阿鹅、萌番薯。
      * --坚果炒货:夏威夷果、核桃、腰果、杏仁、巴旦木、松子、瓜子、腰果、开心果、核桃      ...以形式提：礼包、礼盒
      * --牛奶：常温/低温牛奶
      */
    /**
      * 农产品大单品撰写品类如下：
      * 水果：苹果、
      * 淘宝天猫：50050732
      * 盒马：tbCategoryId 126198303 126198306 126198308
      * 苏宁：503926
      * 京东：13554
      * 拼多多：8269  2447  171  7718  苹果
      *
      * 橙子/橘子、
      * 淘宝天猫：
      * categoryId 50280002 123220003
      * 盒马：
      * categoryId txd_10000040 title包含橙、橘
      * subCategoryId 127532102
      * tbCategoryId 127532101
      * 苏宁：
      * subCategoryId 503923 title包含橘
      * 京东：
      * subCategoryId 13555 13567
      * 拼多多：subCategoryId 5122 橙 1802 5123 桔子 165
      *
      * 车厘子/樱桃；
      * 淘宝天猫：
      * categoryId：50050730
      * 盒马：
      * tbCategoryId 126198318
      * 苏宁：detailCategoryName 车厘子
      * 京东：
      * subCategoryId 13557
      * 拼多多：subCategoryId 1212 车厘子樱桃 161 8249
      *
      * 粮油调味：大米、100210203
      * 杂粮；100210205
      * 海鲜：小龙虾、
      * 虾类
      * 淘宝天猫：
      * 50016794
      * 50050706
      * 50050707
      * 123220002
      * title 含小龙虾
      * 盒马：subCategoryId 126198437 txd_10012075 127530162 127530159
      * 127530160
      * 127530161 127526171 title含小龙虾
      * 苏宁：subCategoryId 503931 title含小龙虾
      * 京东：subCategoryId 12242 title含小龙虾
      * 拼多多：subCategoryId 虾类 175 8195 title含小龙虾
      *
      * 蟹类；
      * 淘宝天猫：
      * 50016792
      * 50016800
      * 50050708
      * 50050710
      * 123188006
      * 123200006
      * 123224004
      * 127692021
      * 50050709
      * 123222005
      * 盒马:subCategoryId
      * 127532156
      * 127530102
      * 127528157
      * 127526172
      * 126198448
      * 126198436
      * 苏宁：503934
      * 京东：12243
      * 拼多多：subCategoryId 蟹类 176 8196
      *
      * 茶：红茶	210804
      * 、绿茶	210803
      * 黑茶 拼多多：subCategoryId 6610 6611 6612 5272
      * 盒马：126198766 title含黑茶、茯茶、千两茶、黑砖茶、三尖、青砖茶、藏茶、边茶、安茶、普洱熟茶、六堡茶
      * 其他平台：100210808
      * 休闲食品：蜜饯干果 210105
      */
    //    val years = "2019"
    for(years <- Array("2018","2019")){

    //"1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12" ,"jd"
    for (months <- Array("1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12")) {
      //"taobao","jd","tmall","suning","kaola", "guomei", "dangdang", "jumei", "miya", "womaiwang", "shunfeng", "secoo", "babytree", "beibei","pinduoduo"
      for (platform <- Array("jd","taobao","tmall","hmxs")) {
//        jd_fromes(sc,sqlContext,years,months,platform)
//        suning_fromes(sc,sqlContext,years,months,platform)
        if(platform.equals("jd") && years.equals("2018")) jd_fromes(sc,sqlContext,years,months,platform)
        else if(platform.equals("suning")) suning_fromes(sc,sqlContext,years,months,platform)
        else if(platform.equals("taobao") |platform.equals("tmall")) tmtaobao_fromes(sc,sqlContext,years,months,platform)
        else if(platform.equals("hmxs") && years.equals("2019")) hmxs_fromes(sc,sqlContext,years,months,platform)
        else if(platform.equals("pinduoduo") && years.concat(months)!="20181" ) pinduoduo_fromes(sc,sqlContext,years,months,platform)
           //普洱


      }
      }
    }



    sc.stop()

  }

  def jd_fromes(sc:SparkContext,sqlContext:SQLContext,years:String,months:String,platform:String):Unit = {

    var frame = sqlContext.emptyDataFrame
    //"小米", "华为", "荣耀", "VIVO", "OPPO", "三星", "魅族","苹果"
    try {


      //xiaomi_demojd/xmjd_2018_12  ${years}_${platform}/${platform}_${years}_${months}
      val data1 = sc.esJsonRDD(s"${years}_${platform}/${platform}_${years}_${months}",
        s"""
           |{
           |  "size": 0
           |  , "query": {
           |    "bool": {
           |      "should": [
           |        {
           |          "terms": {
           |            "subCategoryId": [
           |              "13554",
           |              "13555",
           |              "13567",
           |              "13557",
           |              "12242",
           |              "12243"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "thirdCategoryId": [
           |              "100210203",
           |              "100210205",
           |              "100210804",
           |              "100210803",
           |              "100210808",
           |              "100210105"
           |            ]
           |          }
           |        }
           |      ]
           |    }
           |  },"aggs": {
           |    "NAME": {
           |      "terms": {
           |        "field": "title.keyword",
           |        "size": 10
           |      }
           |    }
           |  }
           |}
      """.stripMargin).values
      val value = data1.map(line => {
        val nObject = JSON.parseObject(line)
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea", "-1").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val detailCategoryName = nObject.getOrDefault("detailCategoryName", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("detailCategoryName", detailCategoryName)
        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("title", title)
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount)
        nObject.put("originProduction", originProduction)
        nObject.put("shopId", shopId)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandValueId", brandValueId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("platformId", platformId)
        nObject.put("Base_Info", Base_Info)
        nObject.put("categoryId", categoryId)
        nObject.toJSONString
      })
      val df1 = sqlContext.read.json(value).selectExpr("good_id","Base_Info","title","priceText","sellCount"
        ,"salesAmount","originProduction","shopId","province",
        "city","district","regional_ID","brandName_cn","brandValueId","timeStamp","platformId","subCategoryId as categoryId","detailCategoryName",
        "thirdCategoryId","secondCategoryId","fourthCategoryId","is_oversea")



      frame = df1
    }catch {
      case ex: Exception => frame
    }


    frame
      .repartition(5).write.orc(s"s3a://o2o-dataproces-group/han_congcong/Dadanpin/source_fourthCategoryId0403/${years}/${platform}/${months}")

  }

  def suning_fromes(sc:SparkContext,sqlContext:SQLContext,years:String,months:String,platform:String):Unit = {

    var frame = sqlContext.emptyDataFrame
    //"小米", "华为", "荣耀", "VIVO", "OPPO", "三星", "魅族","苹果"
    try {


      //xiaomi_demojd/xmjd_2018_12  ${years}_${platform}/${platform}_${years}_${months}
      val data1 = sc.esJsonRDD(s"${years}_${platform}/${platform}_${years}_${months}",
        s"""
           |{
           |  "size": 0
           |  , "query": {
           |    "bool": {
           |      "should": [
           |        {
           |          "terms": {
           |            "subCategoryId": [
           |              "503926",
           |              "503923",
           |              "503931",
           |              "503934"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "detailCategoryName.keyword": [
           |              "车厘子"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "thirdCategoryId": [
           |              "100210203",
           |              "100210205",
           |              "100210804",
           |              "100210803",
           |              "100210808",
           |              "100210105"
           |            ]
           |          }
           |        }
           |      ]
           |    }
           |  },"aggs": {
           |    "NAME": {
           |      "terms": {
           |        "field": "title.keyword",
           |        "size": 10
           |      }
           |    }
           |  }
           |}
      """.stripMargin).values
      val value = data1.map(line => {
        val nObject = JSON.parseObject(line)
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea", "-1").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("title", title)
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount)
        nObject.put("originProduction", originProduction)
        nObject.put("shopId", shopId)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandValueId", brandValueId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("platformId", platformId)
        nObject.put("Base_Info", Base_Info)
        nObject.put("categoryId", categoryId)
        nObject.toJSONString
      })
      val df1 = sqlContext.read.json(value).selectExpr("good_id","Base_Info","title","priceText","sellCount"
        ,"salesAmount","originProduction","shopId","province",
        "city","district","regional_ID","brandName_cn","brandValueId","timeStamp","platformId","subCategoryId as categoryId","detailCategoryName",
        "thirdCategoryId","secondCategoryId","fourthCategoryId","is_oversea")

      frame = df1
    }catch {
      case ex: Exception => frame
    }


    frame
      .repartition(5).write.orc(s"s3a://o2o-dataproces-group/han_congcong/Dadanpin/source_fourthCategoryId0403/${years}/${platform}/${months}")

  }

  def pinduoduo_fromes(sc:SparkContext,sqlContext:SQLContext,years:String,months:String,platform:String):Unit = {

    var frame = sqlContext.emptyDataFrame
    //"小米", "华为", "荣耀", "VIVO", "OPPO", "三星", "魅族","苹果"
    try {
      //xiaomi_demojd/xmjd_2018_12  ${years}_${platform}/${platform}_${years}_${months}
      val data1 = sc.esJsonRDD(s"${years}_${platform}/${platform}_${years}_${months}",
        s"""
           |{
           |  "size": 0
           |  , "query": {
           |    "bool": {
           |      "should": [
           |        {
           |          "terms": {
           |            "subCategoryId": [
           |              "8269",
           |              "2447",
           |              "171",
           |              "7718",
           |              "5122",
           |              "橙",
           |              "1802",
           |              "5123",
           |              "桔子",
           |              "165",
           |              "1212",
           |              "车厘子樱桃",
           |              "161",
           |              "8249",
           |              "虾类",
           |              "175",
           |              "8195",
           |              "蟹类",
           |              "176",
           |              "8196",
           |              "6610",
           |              "6611",
           |              "6612",
           |              "5272"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "thirdCategoryId": [
           |              "100210203",
           |              "100210205",
           |              "100210804",
           |              "100210803",
           |              "100210105"
           |            ]
           |          }
           |        }
           |      ]
           |    }
           |  },"aggs": {
           |    "NAME": {
           |      "terms": {
           |        "field": "title.keyword",
           |        "size": 10
           |      }
           |    }
           |  }
           |}
      """.stripMargin).values
      val value = data1.map(line => {
        val nObject = JSON.parseObject(line)
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea", "false").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("title", title)
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount)
        nObject.put("originProduction", originProduction)
        nObject.put("shopId", shopId)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandValueId", brandValueId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("platformId", platformId)
        nObject.put("Base_Info", Base_Info)
        nObject.put("categoryId", categoryId)
        nObject.toJSONString
      })
      val df1 = sqlContext.read.json(value).selectExpr("good_id","'-1' as Base_Info","title","priceText","sellCount"
        ,"salesAmount","originProduction","shopId","province",
        "city","district","regional_ID","'-1' as brandName_cn","'-1' as brandValueId","timeStamp","platformId","subCategoryId as categoryId","'-1' as detailCategoryName",
        "thirdCategoryId","secondCategoryId","fourthCategoryId","is_oversea")
      frame = df1
    }catch {
      case ex: Exception => frame
    }


    frame
      .repartition(5).write.orc(s"s3a://o2o-dataproces-group/han_congcong/Dadanpin/source_fourthCategoryId0403/${years}/${platform}/${months}")

  }

  def tmtaobao_fromes(sc:SparkContext,sqlContext:SQLContext,years:String,months:String,platform:String):Unit = {

    var frame = sqlContext.emptyDataFrame
    //"小米", "华为", "荣耀", "VIVO", "OPPO", "三星", "魅族","苹果"
    try {


      val data1 = sc.esJsonRDD(s"${years}_${platform}/${platform}_${years}_${months}",
        s"""
           |{
           |  "size": 0
           |  , "query": {
           |    "bool": {
           |      "should": [
           |        {
           |          "terms": {
           |            "categoryId": [
           |              "50050732",
           |              "50280002",
           |              "123220003",
           |              "50050730",
           |              "50016794",
           |              "50050706",
           |              "50050707",
           |              "123220002",
           |              "50016792",
           |"50016800",
           |"50050708",
           |"50050710",
           |"123188006",
           |"123200006",
           |"123224004",
           |"127692021",
           |"50050709",
           |"123222005"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "thirdCategoryId": [
           |              "100210203",
           |              "100210205",
           |              "100210804",
           |              "100210803",
           |              "100210808",
           |              "100210105"
           |            ]
           |          }
           |        }
           |      ]
           |    }
           |  },"aggs": {
           |    "NAME": {
           |      "terms": {
           |        "field": "title.keyword",
           |        "size": 10
           |      }
           |    }
           |  }
           |}
      """.stripMargin).values
      val value = data1.map(line => {
        val nObject = JSON.parseObject(line)
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val detailCategoryName = nObject.getOrDefault("detailCategoryName", "-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea", "-1").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("detailCategoryName", detailCategoryName)
        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("title", title)
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount)
        nObject.put("originProduction", originProduction)
        nObject.put("shopId", shopId)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandValueId", brandValueId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("platformId", platformId)
        nObject.put("Base_Info", Base_Info)
        nObject.put("categoryId", categoryId)
        nObject.toJSONString
      })

      val df3 = sqlContext.read.json(value).selectExpr("good_id","Base_Info","title","priceText","sellCount"
        ,"salesAmount","originProduction","shopId","province",
        "city","district","regional_ID","brandName_cn","brandValueId","timeStamp","platformId","categoryId","'-1' as detailCategoryName",
        "thirdCategoryId","secondCategoryId","fourthCategoryId","is_oversea")

      frame = df3
    }catch {
      case ex: Exception => frame
    }


    frame
      .repartition(5).write.orc(s"s3a://o2o-dataproces-group/han_congcong/Dadanpin/source_fourthCategoryId0403/${years}/${platform}/${months}")

  }

  def hmxs_fromes(sc:SparkContext,sqlContext:SQLContext,years:String,months:String,platform:String):Unit = {

    var frame = sqlContext.emptyDataFrame
    //"小米", "华为", "荣耀", "VIVO", "OPPO", "三星", "魅族","苹果"
    try {


      //xiaomi_demojd/xmjd_2018_12  ${years}_${platform}/${platform}_${years}_${months}
      val data1 = sc.esJsonRDD(s"${years}_${platform}/${platform}_${years}_${months}",
        s"""
           |{
           |  "size": 0
           |  , "query": {
           |    "bool": {
           |      "should": [
           |        {
           |          "terms": {
           |            "tbCategoryId": [
           |              "126198303",
           |              "126198306",
           |              "126198308",
           |              "127532101",
           |              "126198318",
           |              "127532102",
           |              "126198437",
           |              "txd_10012075",
           |              "127530162",
           |              "127530159",
           |              "127530160",
           |              "127530161",
           |              "127526171",
           |              "127532156",
           |"127530102",
           |"127528157",
           |"127526172",
           |"126198448",
           |"126198436",
           |"126198766"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "categoryId": [
           |              "txd_10000040"
           |            ]
           |          }
           |        },{
           |          "terms": {
           |            "thirdCategoryId": [
           |              "100210203",
           |              "100210205",
           |              "100210804",
           |              "100210803",
           |              "100210105"
           |            ]
           |          }
           |        }
           |      ]
           |    }
           |  },"aggs": {
           |    "NAME": {
           |      "terms": {
           |        "field": "tbCategoryId",
           |        "size": 10
           |      }
           |    }
           |  }
           |}
      """.stripMargin).values

      val value = data1.map(line => {
        val nObject = JSON.parseObject(line)
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val detailCategoryName = nObject.getOrDefault("detailCategoryName", "-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea", "false").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("detailCategoryName", detailCategoryName)
        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("title", title)
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", salesAmount)
        nObject.put("originProduction", originProduction)
        nObject.put("shopId", shopId)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandValueId", brandValueId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("platformId", platformId)
        nObject.put("Base_Info", Base_Info)
        nObject.put("categoryId", categoryId)
        nObject.toJSONString
      })

      val df1 = sqlContext.read.json(value).selectExpr("good_id","Base_Info","title","priceText","sellCount"
        ,"salesAmount","originCountryName as originProduction","shopId","province",
        "city","district","regional_ID","brandName_cn","brandValueId","timeStamp","platformId","tbCategoryId","categoryId","subCategoryId","'-1' as detailCategoryName",
        "thirdCategoryId","secondCategoryId","fourthCategoryId","'false' as is_oversea")

      frame = df1
    }catch {
      case ex: Exception => frame
    }


    frame
      .repartition(5).write.orc(s"s3a://o2o-dataproces-group/han_congcong/Dadanpin/source_fourthCategoryId0403/${years}/${platform}/${months}")

  }

}
